package com.stone.im.config;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.stone.im.common.Constants;
import com.stone.im.common.SnowflakeIdWorker;
import com.stone.im.dispatcher.KafkaMenager;
import com.stone.im.entity.MessageReceiveEntity;
import com.stone.im.entity.MessageSendEntity;
import com.stone.im.service.MessageReceiveService;
import com.stone.im.service.MessageSendService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.sql.Wrapper;
import java.util.Arrays;
import java.util.Date;

@Component
public class KafkaConfig {

    @Autowired
    private static MessageSendService messageSendService;

    @Autowired
    private static MessageReceiveService messageReceiveService;

    public void test(){
        KafkaMenager instance = KafkaMenager.getInstance();
        KafkaConsumer consumer = instance.getConsumer();
        consumer.subscribe(Arrays.asList("send_message","push_message_response"));

        while (true){
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String,String> recode:records){
                String topic = recode.topic();

                JSONObject messageJson = JSONObject.parseObject(recode.value());

                if (topic.equals("send_message")){
                    //将数据写入mysql
                    String messageId = stareMessage(messageJson);
                    //返回消息发送的响应
                    sendMessageResponse(messageJson,messageId);
                    //发送消息推送
                    sendMessagePush(messageJson,messageId);
                }else if (topic.equals("push_message_response")){
                    processPushMessageResponse(messageJson);
                }
            }
        }

    }

    /**
     * 处理推送消息的响应
     * @param messageJson
     */
    private void processPushMessageResponse(JSONObject messageJson) {
        String messageId = messageJson.getString("messageId");
        MessageReceiveEntity messageReceiveEntity = new MessageReceiveEntity();
        messageReceiveEntity.setMessageId(messageId);
        messageReceiveEntity.setIsDelivered(1);

        QueryWrapper<MessageReceiveEntity> wrapper = new QueryWrapper<>();
        wrapper.eq("messageId",messageId);

        messageReceiveService.update(messageReceiveEntity,wrapper);
    }

    private static String stareMessage(JSONObject messageJson){

        SnowflakeIdWorker idWorker = new SnowflakeIdWorker(0, 0);

        String senderId = messageJson.getString("senderId");
        String receiverId = messageJson.getString("receiverId");
        String context = messageJson.getString("context");
        Integer requestType = messageJson.getInteger("requestType");
        Integer sequence = messageJson.getInteger("sequence");
        String messageId = String.valueOf(idWorker.nextId());

        MessageSendEntity messageSendEntity = new MessageSendEntity();
        messageSendEntity.setMessageId(messageId);
        messageSendEntity.setSenderId(senderId);
        messageSendEntity.setReceiverId(receiverId);
        messageSendEntity.setContent(context);
        messageSendEntity.setRequestType(requestType);
        messageSendEntity.setSequence(sequence);
        messageSendEntity.setSendTime(new Date());
        messageSendEntity.setMessageType(1);

        messageSendService.save(messageSendEntity);

        MessageReceiveEntity messageReceiveEntity = new MessageReceiveEntity();
        messageReceiveEntity.setMessageId(messageId);
        messageReceiveEntity.setSenderId(senderId);
        messageReceiveEntity.setReceiverId(receiverId);
        messageReceiveEntity.setContent(context);
        messageReceiveEntity.setRequestType(requestType);
        messageReceiveEntity.setSequence(sequence);
        messageReceiveEntity.setSendTime(new Date());
        messageReceiveEntity.setMessageType(1);
        messageReceiveEntity.setIsDelivered(0);
        messageReceiveService.save(messageReceiveEntity);
        
        return messageId;
    }

    /**
     * 消息推送
     * @param message
     */
    public void sendMessagePush(JSONObject message,String messageId){
        KafkaMenager instance = KafkaMenager.getInstance();
        KafkaProducer producer = instance.getProducer();

        message.put("messageId",messageId);
        message.put("timestamp",System.currentTimeMillis());
        message.put("requestType",Constants.REQUEST_TYPE_PUSH_MESSAGE);

        ProducerRecord<String, String> record = new ProducerRecord<>(
                "send_message_response", message.toJSONString()
        );
        producer.send(record);
    }

    private void sendMessageResponse(JSONObject messageJson,String messageId){

        Integer requestType = messageJson.getInteger("requestType");
        String senderId = messageJson.getString("senderId");
        String receiverId = messageJson.getString("receiverId");
        String gatewayChannelId = messageJson.getString("gatewayChannelId");

        KafkaMenager instance = KafkaMenager.getInstance();
        KafkaProducer producer = instance.getProducer();
        
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("messageId",messageId);
        jsonObject.put("timestamp",System.currentTimeMillis());
        jsonObject.put("gatewayChannelId",gatewayChannelId);
        jsonObject.put("requestType",requestType);
        jsonObject.put("senderId",senderId);
        jsonObject.put("receiverId",receiverId);
        ProducerRecord<String, String> record = new ProducerRecord<>(
                "push_message", jsonObject.toJSONString()
        );
        producer.send(record);

    }
}
